package org.leafDomain.combine.Impl;

import org.apache.hadoop.hbase.client.Put;
import org.apache.log4j.Logger;
import org.leafDomain.combine.DataSourceToHBase;
import org.leafDomain.hbase.HBase;
import org.leafDomain.rdb.RDB;

import java.util.List;

public class RDBToHBaseImpl implements DataSourceToHBase<RDB>{
    
    // Logger在哪个类里面，就用哪个类的Class信息，方便记录到日志的时候，能把错误是哪个类的记录下来
    private static final Logger loggerRH = Logger.getLogger(RDBToHBaseImpl.class);
    private HBase hbase;
    private RDB rdb;

    @Override
    public void setSource(RDB rdb) {
        this.rdb = rdb;
    }

    @Override
    public void setHBase(HBase hbase) {
        this.hbase = hbase;
    }

    @Override
    public void startTransfer() {
        try {
            rdb.init();
            loggerRH.info("RDB 初始化成功");
            hbase.init();
            loggerRH.info("HBase 初始化成功");

            loggerRH.info("数据从 RDB 迁移至 HBase 开始...");
            int count = 0;
            while (rdb.hashNextBatch()){
                final List<Put> batch = rdb.nextBatch();
                hbase.putBatch(batch);
                loggerRH.info(String.format("第 %d 批：%d 条数据插入成功", ++count, batch.size()));
            }
            loggerRH.info("数据从 RDB 迁移至 HBase 结束...");
        }catch (Exception e){
            loggerRH.error("将 RDB 数据批量迁移至 HBase 异常：",e);
        } finally {
            // 先开的后释放后开的先释放
            hbase.close();
            rdb.close();
        }
    }
}
